#include "cgzCMongo.h"
#include "cgzCThreadPool.h"

mongocxx::instance inst{};

cgzCMongoReply::cgzCMongoReply(cgzCMongoCmd cmd, long cookid)
    : m_cmd(cmd),
      m_err(false),
      m_cookid(cookid)
{
}

cgzCMongoReply::~cgzCMongoReply()
{
}

cgzCMongoExec::cgzCMongoExec(const cgzCMongoCmd cmd, const char *collection, bsoncxx::builder::stream::document &doc, const int cookid)
    : cgzCMongoReply(cmd, cookid),
      m_collection(collection),
      m_doc(std::move(doc)),
      m_ret(false)
{
}

cgzCMongoExec::cgzCMongoExec(const cgzCMongoCmd cmd, const char *collection, std::vector<bsoncxx::document::value> &docs, const int cookid)
    : cgzCMongoReply(cmd, cookid),
      m_collection(collection),
      m_docs(std::move(docs)),
      m_ret(false)
{
}

cgzCMongoExec::~cgzCMongoExec()
{
}

cgzCMongoResult::cgzCMongoResult(const cgzCMongoCmd cmd, const int cookid, bsoncxx::document::value &result)
    : cgzCMongoReply(cmd, cookid),
      m_result(std::move(result))
{
}

cgzCMongoResult::~cgzCMongoResult()
{
}

cgzCMongoResultMany::cgzCMongoResultMany(const cgzCMongoCmd cmd, const int cookid)
    : cgzCMongoReply(cmd, cookid)
{
}

cgzCMongoResultMany::~cgzCMongoResultMany()
{
}

void cgzMongoCloseCheck(mongocxx::client **c)
{
    if (*c)
    {
        delete (*c);
    }
    *c = nullptr;
}

cgzCMongo::cgzCMongo(
    cgzCThreadPool *pool,
    const char *name,
    const char *ip,
    const char *user,
    const char *password,
    const int port,
    const char *db,
    const int cid)
    : cgzITask(pool, name, cgzETask::CMongo),
      m_ip(ip),
      m_user(user),
      m_password(password),
      m_db(db),
      m_firstAuth(false),
      m_ping(false),
      m_port(port),
      m_cid(cid),
      m_Mongo(nullptr)

{
    // 尝试连接
    this->connect();
    // 注册服务
    service();
}

cgzCMongo::~cgzCMongo()
{
    if (nullptr != m_Mongo)
    {
        delete m_Mongo;
        m_Mongo = nullptr;
    }
}

void cgzCMongo::authret()
{
    m_ping = true;
    if (!m_firstAuth)
    {
        m_firstAuth = true;
        auto node = malloc(sizeof(cgzCMongoReply));
        auto acptC = new (node) cgzCMongoReply(cgzCMongoCmd::AUTH, -1);
        todoret(acptC);
    }
}

bool cgzCMongo::rconnect()
{
    cgzMongoCloseCheck(&m_Mongo);
    return connect();
}

bool cgzCMongo::connect()
{
    std::string addr;
    if (m_user.empty() || m_password.empty())
    {
        addr = "mongodb://" + m_ip + ":" + std::to_string(m_port);
    }
    else
    {
        addr = "mongodb://" + m_user + ":" + m_password + "@" + m_ip + ":" + std::to_string(m_port);
    }

    mongocxx::uri uri(addr);
    m_Mongo = new mongocxx::client(uri);

    try
    {
        // 连接数据
        auto ping_cmd = bsoncxx::builder::basic::make_document(bsoncxx::builder::basic::kvp("ping", 1));
        m_Mongo->database(m_db.c_str()).run_command(ping_cmd.view());

        authret();
        return true;
    }
    catch (const std::exception &e)
    {
        cgz_this_error("Connection failed: {}", e.what());
    }
    return false;
}

void cgzCMongo::todoret(cgzCMongoReply *result)
{
    result->origin = getId();
    result->target = m_cid;
    retTask(result);
}

bool cgzCMongo::run(int64_t nms)
{
    bool no = false;
    while (!m_runTask.empty())
    {
        bool ex = false;
        auto task = static_cast<cgzCMongoExec *>(m_runTask.front());
        if (no)
        {
            // 无法重连成功
        }
        else if (!m_ping)
        {
            if (rconnect())
            {
                // 重新连接成功了
                continue;
            }
            else
            {
                no = true;
                // 无法重连成了
            }
        }
        else
        {
            try
            {
                switch (task->m_cmd)
                {
                case cgzCMongoCmd::NONE:
                    none(task);
                    break;
                case cgzCMongoCmd::INSERT:
                    add(task);
                    break;
                case cgzCMongoCmd::INSERTS:
                    adds(task);
                    break;
                case cgzCMongoCmd::DELETE_ONE:
                    del(task);
                    break;
                case cgzCMongoCmd::DELETE:
                    dels(task);
                    break;
                case cgzCMongoCmd::FIND_ONE:
                    get(task);
                    break;
                case cgzCMongoCmd::FIND:
                    gets(task);
                    break;
                case cgzCMongoCmd::UPDATE:
                    set(task);
                    break;
                case cgzCMongoCmd::COUNT:
                    num(task);
                    break;
                case cgzCMongoCmd::AGGREGATE:
                    agg(task);
                    break;
                default:
                {
                    task->m_err = true;
                    none(task);
                    cgz_this_error("unknown cmd:{}", (int)task->m_cmd);
                }
                break;
                }
            }
            catch (const std::exception &e)
            {
                ex = true;
                std::cerr << e.what() << '\n';
            }
        }

        // 抛出异常-但是还没有重连失败
        if (ex && (!no))
        {
            // 尝试重连
            m_ping = false;
            continue;
        }

        m_runTask.pop_front();
    }

    return true;
}

void cgzCMongo::add(cgzCMongoExec *exec)
{
    auto db = (*m_Mongo)[m_db.c_str()];
    auto collection = db[exec->m_collection.c_str()];
    auto result = collection.insert_one(exec->m_doc.view());
    if (!result)
    {
        exec->m_err = true;
        cgz_this_error("{}", "Insert failed");
    }
    else
    {
        bsoncxx::oid inserted_id = result->inserted_id().get_oid().value;
        exec->m_id = inserted_id.to_string();
    }
    none(exec);
}

void cgzCMongo::adds(cgzCMongoExec *exec)
{
    // 获取集合引用
    auto db = (*m_Mongo)[m_db.c_str()];
    auto collection = db[exec->m_collection.c_str()];

    // 配置插入选项
    mongocxx::options::insert options;
    options.ordered(true); // 有序插入提高性能
    // 通用兼容写法（适用于大多数版本）
    mongocxx::write_concern wc;
    wc.acknowledge_level(mongocxx::write_concern::level::k_majority); // 设置确认级别
    wc.timeout(std::chrono::seconds(5));                              // 设置超时

    options.write_concern(wc);

    // 执行批量插入
    auto result = collection.insert_many(exec->m_docs, options);

    // 处理结果
    if (!result)
    {
        exec->m_err = true;
        cgz_this_error("{}", "Insert many operation failed with no result");
    }
    none(exec);
}

void cgzCMongo::del(cgzCMongoExec *exec)
{
    auto db = (*m_Mongo)[m_db.c_str()];
    auto collection = db[exec->m_collection.c_str()];
    auto result = collection.delete_one(exec->m_doc.view());
    if (!result)
    {
        exec->m_err = true;
        cgz_this_error("{}", "Delete failed");
    }
    else
    {
        exec->m_delete = result->deleted_count();
        cgz_this_error("Delete count:{}", exec->m_delete);
    }
    none(exec);
}

void cgzCMongo::dels(cgzCMongoExec *exec)
{
    auto db = (*m_Mongo)[m_db.c_str()];
    auto collection = db[exec->m_collection.c_str()];
    auto result = collection.delete_many(exec->m_doc.view());
    if (!result)
    {
        exec->m_err = true;
        cgz_this_error("{}", "Delete many failed");
    }
    else
    {
        exec->m_delete = result->deleted_count();
        cgz_this_error("Delete count:{}", exec->m_delete);
    }
    none(exec);
}

void cgzCMongo::get(cgzCMongoExec *exec)
{
    auto db = (*m_Mongo)[m_db.c_str()];
    auto collection = db[exec->m_collection.c_str()];
    auto result = collection.find_one(exec->m_doc.view());
    if (!result)
    {
        exec->m_err = true;
        cgz_this_error("{}", "Find one failed");
        return;
    }

    auto node = malloc(sizeof(cgzCMongoResult));
    auto retC = new (node) cgzCMongoResult(cgzCMongoCmd::FIND_ONE, exec->m_cookid, *result);
    todoret(static_cast<cgzCMongoReply *>(retC));
}

void cgzCMongo::gets(cgzCMongoExec *exec)
{
    auto db = (*m_Mongo)[m_db.c_str()];
    auto collection = db[exec->m_collection.c_str()];
    // 使用 find 查询多个文档
    auto cursor = collection.find(exec->m_doc.view());

    auto node = malloc(sizeof(cgzCMongoResultMany));
    auto retC = new (node) cgzCMongoResultMany(cgzCMongoCmd::FIND, exec->m_cookid);
    for (auto &&doc : cursor)
    {
        retC->m_result.push_back(bsoncxx::document::value(doc)); // 将每个文档添加到结果集中
    }
    todoret(static_cast<cgzCMongoReply *>(retC));
}

void cgzCMongo::set(cgzCMongoExec *exec)
{
    auto db = (*m_Mongo)[m_db.c_str()];
    auto collection = db[exec->m_collection.c_str()];
    auto result = collection.update_one(exec->m_doc.view(), exec->m_doc.view());
    if (!result)
    {
        exec->m_err = true;
        cgz_this_error("{}", "Update failed");
    }
    return none(exec);
}

void cgzCMongo::num(cgzCMongoExec *exec)
{
    auto db = (*m_Mongo)[m_db.c_str()];
    auto collection = db[exec->m_collection.c_str()];
    auto result = collection.count_documents(exec->m_doc.view());
    if (!result)
    {
        exec->m_err = true;
        cgz_this_error("{}", "Count failed");
    }
    else
    {
        std::cout << "Count: " << result << std::endl;
    }
    return none(exec);
}

void cgzCMongo::agg(cgzCMongoExec *exec)
{
    cgz_this_error("Not support aggregate type:{}", (int)exec->m_cmd);
    none(exec);
}

void cgzCMongo::none(cgzCMongoExec *exec)
{
    if (!exec->m_ret)
    {
        exec->~cgzCMongoExec();
        free(exec);
        return;
    }
    todoret(exec);
}
